package com.study.flink.table

import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.TableEnvironment

/**
  *
  * @author stephen
  * @date 2019-07-22 17:40
  */
object ScalaStreamTableDemo {

  def main(args: Array[String]): Unit = {
    // 1 获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 2 获取输入数据
    import org.apache.flink.api.scala._
    val dataStream = env.socketTextStream("localhost", 9999)


    // 3 Transformation
    val eventStream = dataStream.map(x => {
      val arr = x.split(",")
      SalesLog(arr(0).toLong, arr(1), arr(2), arr(3),arr(4).toDouble)
    })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SalesLog](Time.seconds(0)) {
        override def extractTimestamp(element: SalesLog): Long = element.ts
      })
    val tableEnv = TableEnvironment.getTableEnvironment(env)
    // case class 可以根据其结构直接生成表
    val table = tableEnv.fromDataStream(eventStream)
    //val resultTable:Table = table.window(Tumble over 10000.millis on 'ts as 'tt).groupBy('itemId,'tt ).select( 'itemId, 'itemId.count)
    //val resultStream = resultTable.toRetractStream[(String,Long)]

    // 4 输出
    //resultStream.print()

    // 5 启动任务
    env.execute("Demo")
  }
}

case class SalesLog(ts: Long,
                    transactionId: String,
                    customerId: String,
                    itemId: String,
                    amountPaid: Double)
